package com.supertool.tong.merger.util;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import java.io.IOException;

public class MzSequenceFileRecordReader extends RecordReader<LongWritable, Text> {

    private MzSequenceFile.Reader in;
    //protected Configuration conf;
    private LongWritable key = new LongWritable(0);
    private Text value = new Text();
    private long start;
    private long end;

    public MzSequenceFileRecordReader(Configuration conf, FileSplit split)
            throws IOException {
        Path path = split.getPath();
        FileSystem fs = path.getFileSystem(conf);
        this.in = new MzSequenceFile.Reader(fs, path, fs.getLength(path),conf, conf.get("miaozhen.seq.meta"));
        //this.conf = conf;
        this.start = this.in.getStart();
        this.end = this.in.getEnd();
    }
    public void initialize(InputSplit split, TaskAttemptContext context)
            throws IOException, InterruptedException {
    }
    public Class getValueClass() { return in.getValueClass(); }

    public LongWritable getCurrentKey() {
        return this.key;
    }

    public Text getCurrentValue() {
        return this.value;
    }
    public  boolean nextKeyValue() throws IOException {

        return in.next((Object)this.value);

    }

    public float getProgress() throws IOException {
        if (end == start) {
            return 0.0f;
        } else {
            return Math.min(1.0f, (in.getPos() - start) / (float)(end - start));
        }
    }

    public long getPos() throws IOException {
        return in.getPos();
    }

    protected void seek(long pos) throws IOException {
        throw new IOException("MzSequenceFile does not support seek");
    }
    public void close() throws IOException { in.close(); }

}
